package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/segmentio/kafka-go"
)

// Kafka 配置
const (
	kafkaAddress  = "127.0.0.1:9092"
	topicName     = "test"
	consumergroup = "consumer-group-id"
)

var r *kafka.Reader

func initKafkaReader() {
	r = kafka.NewReader(kafka.ReaderConfig{
		Brokers:        []string{kafkaAddress}, // 设置Kafka代理地址列表（配置）
		Topic:          topicName,              // 指定要读取的主题（配置）
		GroupID:        consumergroup,          // 设置消费者组ID，支持消费组管理（配置）
		MinBytes:       1,                      // 设置单次读取的最小字节数（选填）
		MaxBytes:       10e6,                   // 设置单次读取的最大字节数为10MB（保持）
		MaxWait:        10 * time.Second,       // 获取批处理时等待新数据的最大时间（选填）
		StartOffset:    kafka.LastOffset,       // 指定读取的起始偏移量，从最新的消息开始（保持，默认kafka.FirstOffset）
		CommitInterval: 0,                      // 自动提交偏移量的时间间隔，为0时为同步提交（保持，每秒time.Second）
	})

	// 测试连接 Kafka
	conn, err := kafka.Dial("tcp", kafkaAddress)
	if err != nil {
		fmt.Printf("连接Kafka broker失败: %v", err)
		return
	}
	defer conn.Close()

	// 确保 topic 存在
	_, err = conn.ReadPartitions(topicName)
	if err != nil {
		fmt.Printf("找不到topic %s: %v", topicName, err)
		r.Close()
		return
	}
	fmt.Println("连接Kafka broker成功")

}

func main() {
	initKafkaReader()
	fmt.Println("开始消费消息...")
	for {
		m, err := r.ReadMessage(context.Background())
		if err != nil {
			log.Printf("读取消息时出错: %v", err)
			break
		}

		fmt.Printf("主题/分区/偏移量 %v/%v/%v: %s = %s\n", m.Topic, m.Partition, m.Offset, string(m.Key), string(m.Value))
		fmt.Println("消息写入时间", m.Time)
	}

	if err := r.Close(); err != nil {
		log.Fatal("failed to close reader:", err)
	}
}
